package com.pxene.dmp.reducer;

import java.io.IOException;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class GetUserInfoReducer extends Reducer<Text, LongWritable, LongWritable, Text> {

	private LongWritable resultKey = new LongWritable();
	private Text resultVal = new Text();

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
	}

	@Override
	protected void reduce(Text key, Iterable<LongWritable> values, Context context)
			throws IOException, InterruptedException {
		String[] columns = key.toString().split("\\|");
		if (columns.length != 11) {
			return;
		}
		String adid = columns[0];
		String type = columns[1];
		String imsi = columns[2];
		String mdn = columns[3];
		String meid = columns[4];
		String device = columns[5];
		String os = columns[6];
		String domainName = columns[7];
		String province = columns[8];
		String dataType = columns[9];
		long sum = Long.parseLong(columns[10]);
		
		long count = 0l;
		for (LongWritable value : values) {
			count += value.get();
		}
		sum = sum + count;
		
		resultKey.set(new Date().getTime());
		resultVal.set(adid + "|" + type + "|" + imsi + "|" + mdn + "|" + meid + "|" + device 
				+ "|" + os + "|" + domainName + "|" + sum + "|" + province + "|" + dataType);
		context.write(resultKey, resultVal);
	}

}
